package edu.purdue.cs.aggr.accumulators;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.List;

import com.google.common.collect.MinMaxPriorityQueue;

import edu.purdue.cs.aggr.AbstractAccumulator;
import edu.purdue.cs.aggr.Accumulator;

public class TopKAccumulator extends AbstractAccumulator implements Accumulator<Long, Long[]> {

	private int K;
	private Long[] tempTopK;
	private int numElements = 0;
	
	public TopKAccumulator(String sourceId) {
	  super(sourceId);
	}
	
	public Long[] getTempTopK() {
		return tempTopK;
	}

	public void setTempTopK(Long[] tempTopK) {
		this.tempTopK = tempTopK;
	}

	public int getNumElements() {
		return numElements;
	}

	public void setNumElements(int numElements) {
		this.numElements = numElements;
	}

	public int getK() {
		return K;
	}

	public void setK(int k) {
		K = k;
	}
	
	public TopKAccumulator(int K, String sourceId) {
	  super(sourceId);
		this.K = K;
		System.out.println("K is:" + K);
		tempTopK = new Long[K];
	}
	
	@Override
	public void initialize(Long input) {
	}

	@Override
	public void addValue(Long input) {
		if (numElements < K) {
			shiftAndAddElement(input);
		} else if (input > tempTopK[0]) {
			removeLowestAndInsert(input);
		}
	}

	private void removeLowestAndInsert(Long input) {
		int insertPos = getInsertPos(input);
		for (int i=0; i<insertPos; i++) {
			tempTopK[i] = tempTopK[i+1];
		}
		tempTopK[insertPos] = input;
		
	}

	private void shiftAndAddElement(Long input) {
		int insertPos = getInsertPos(input);
		for (int i=numElements-1; i>=insertPos; i--) {
			tempTopK[i+1] = tempTopK[i];
		}
		tempTopK[insertPos] = input;
		numElements++;
	}
	
	private int getInsertPos(Long input) {
		int insertPos = 0;
		for (int i=0; i<numElements; i++) {
			if (tempTopK[i] > input) {
				insertPos = i;
			}
		}
		return insertPos;
	}

	@Override
	public Long[] getAccumulation() {
		System.out.println("Total added elements :" + numElements);
		return tempTopK;
	}

	@Override
	public void addAccumulators(List<Accumulator<Long, Long[]>> accumulators) {
		//Merge the two sorted arrays
		
		MinMaxPriorityQueue<Long> priorityQueue = MinMaxPriorityQueue.create();
		
		Long lastValue = null;
		int uniqueValues = 0;
		for (Long l : tempTopK) {
			if (l!=null) {
			  if (lastValue != null && lastValue == l)
				  continue;
			  priorityQueue.add(l);
			}
		}
		
		for (Accumulator<Long, Long[]> acc : accumulators) {
			Long[] tempK = acc.getAccumulation();
			for (Long l : tempK) {
				if (l!=null) {
			      if (lastValue != null && lastValue == l)
				    continue;
				  priorityQueue.add(l);
				}
			}
		}
		
		int i;
		boolean broken = false;
		for (i=tempTopK.length-1; i>=0; i--) {
			Long next = priorityQueue.pollLast();
			if (next==null) {
				broken = true;
				break;
			}
			tempTopK[i] = next;
		}
		
		if (broken) {
			System.out.println("Broken. New list size : " + (tempTopK.length - i));
		} else {
			System.out.println("Not Broken. New list size : " + (tempTopK.length));
		}
		
		
		//TODO
		
		
		
		
//		Long[] tempTopK1 = accumulator.getAccumulation();
//		if (tempTopK1==null) {
//			System.out.println("ERROR : tempTopK1 is null");
//		}
//		Long[] tempTopK2 = tempTopK;
//		if (tempTopK2==null) {
//			System.out.println("ERROR : tempTopK2 is null");
//		}
//		
//		Long[] tempTopK = new Long[K]; //TODO Assuming there to be K unique elements
//		int tempK1 = K-1;
//		int tempK2 = K-1;
//		long previous = -1;
//		for (int i=K-1; i>=0; i--) {
//			if (tempTopK1[tempK1]!=null && tempTopK2[tempK2]==null) {
//				if (previous > 0 && tempTopK1[tempK1]==previous) {
//					tempK1--;
//					continue;
//				}
//				tempTopK[i] = tempTopK1[tempK1];
//				tempK1--;
//		    } else if (tempTopK2[tempK2]!=null && tempTopK1[tempK1]==null) {
//				if (previous > 0 && tempTopK2[tempK2]==previous) {
//					tempK2--;
//					continue;
//				}
//				tempTopK[i] = tempTopK2[tempK2];
//				tempK2--;
//		    } else if (tempTopK2[tempK2]==null && tempTopK1[tempK1]==null) {
//				tempK1--;
//				tempK2--;
//		    } else if (tempTopK1[tempK1] > tempTopK2[tempK2]) {
//				if (previous > 0 && tempTopK1[tempK1]==previous) {
//					tempK1--;
//					continue;
//				}
//				tempTopK[i] = tempTopK1[tempK1];
//				tempK1--;
//			} else {
//				if (previous > 0 && tempTopK2[tempK2]==previous) {
//					tempK2--;
//					continue;
//				}
//				tempTopK[i] = tempTopK2[tempK2];
//				tempK2--;
//			}
//		}
	}

	@Override
	public void readExternal(ObjectInput in) throws IOException,
			ClassNotFoundException {
	  super.readExternal(in);
		K = in.readInt();
		tempTopK = (Long[]) in.readObject();
		numElements = in.readInt();
		
	}

	@Override
	public void writeExternal(ObjectOutput out) throws IOException {
	  super.writeExternal(out);
		out.writeInt(K);
		out.writeObject(tempTopK);
		out.writeInt(numElements);
	}
}
